package com.zyx.flinkcdc;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @author zyx
 * @since 2023/10/31 15:15
 */
public class CdcDemo {

    private static final String HOST = "10.1.207.12";
    private static final int PORT = 3306;
    private static final String USERNAME = "root";
    private static final String PASSWORD = "<PASSWORD>";
    private static final String DATABASES = "db_admin";
    private static final String TABLES = "db_admin.t_cdc_test";

    public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        // 自定义解释器
        // 设置解释器及解释器参数
        debeziumProperties.put("converters", "mydebeziumconverter");
        debeziumProperties.put("mydebeziumconverter.type", "com.zyx.flinkcdc.utils.DebeziumConverter");
        debeziumProperties.put("mydebeziumconverter.database.type", "mysql");
        // 自定义格式，可选
        debeziumProperties.put("mydebeziumconverter.format.datetime", "yyyy-MM-dd HH:mm:ss");
        debeziumProperties.put("mydebeziumconverter.format.date", "yyyy-MM-dd");
        debeziumProperties.put("mydebeziumconverter.format.time", "HH:mm:ss");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname(HOST)
            .port(PORT)
            // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
            .databaseList(DATABASES)
            // set captured table
            .tableList(TABLES)
            .username(USERNAME)
            .password(PASSWORD)
            .startupOptions(StartupOptions.initial())
            .serverTimeZone("Asia/Shanghai")
            .debeziumProperties(debeziumProperties)
            // converts SourceRecord to JSON String
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        // env.enableCheckpointing(3000);

        env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
            // set 4 parallel source tasks
            .setParallelism(4)
            // use parallelism 1 for sink to keep message ordering
            .print()
            .setParallelism(1);
        env.execute("Print MySQL Snapshot + Binlog");
    }

}
